package service

import (
	"fmt"
	"gitee.com/sunny-zhangqiang/zcache_new/comm"
	"github.com/HuKeping/rbtree"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

var (
	cm *chunkMerge
)

type chunkMergeData struct {
	fragmentSize  int64
	fragmentIndex int64
	mergeDone     string
	err           error
}

type chunkMergeNode struct {
	key              string
	clientToken      string
	chunkMergeToken  string
	updateExpireTime int64
	expireTime       int64
	refer            int64
	memAddress       string
	chunkMergeData
	readers   []chan chunkMergeData
	fragments [][]byte
}

func (node *chunkMergeNode) Less(than rbtree.Item) bool {
	thanNode := than.(*chunkMergeNode)

	if node.expireTime < thanNode.expireTime {
		return true
	}

	if node.expireTime > thanNode.expireTime {
		return false
	}

	return node.memAddress < thanNode.memAddress
}

type oneClientChunkMerge struct {
	sync.Mutex
	// map[clientToken] --> map[sourceKey+sourceVersion]*chunkMergeNode
	m map[string]map[string]*chunkMergeNode
}

func (occm *oneClientChunkMerge) insert(node *chunkMergeNode) {
	occm.Lock()
	defer occm.Unlock()

	ctcm := occm.m[node.clientToken]
	if ctcm == nil {
		ctcm = make(map[string]*chunkMergeNode)
		occm.m[node.clientToken] = ctcm
	}

	ctcm[node.key] = node
}

func (occm *oneClientChunkMerge) delete(node *chunkMergeNode) {
	occm.Lock()
	defer occm.Unlock()

	ctcm := occm.m[node.clientToken]
	if ctcm == nil {
		return
	}

	delete(ctcm, node.key)
}

type oneChunkMerge struct {
	sync.Mutex
	m       map[string]*chunkMergeNode
	expireM *rbtree.Rbtree
}

type chunkMerge struct {
	ocms  [slot]oneChunkMerge
	occms [slot]oneClientChunkMerge
}

func (cm *chunkMerge) getOneChunkMerge(key string) *oneChunkMerge {
	hash := comm.FnvHash(comm.String2Bytes(key))
	return &cm.ocms[hash%slot]
}

func (cm *chunkMerge) getOneClientChunkMerge(clientToken string) *oneClientChunkMerge {
	hash := comm.FnvHash(comm.String2Bytes(clientToken))
	return &cm.occms[hash%slot]
}

func (cm *chunkMerge) insert(key string, clientToken string,
	cmData chunkMergeData, chunkExpireTimeout int64) (chunkMergeToken string) {

	ocm := cm.getOneChunkMerge(key)
	occm := cm.getOneClientChunkMerge(clientToken)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node != nil {
		delete(ocm.m, node.key)
		ocm.expireM.Delete(node)
		occm.delete(node)
	}

	expireTime := time.Now().UnixMilli() + chunkExpireTimeout
	node = &chunkMergeNode{
		key:              key,
		clientToken:      clientToken,
		updateExpireTime: expireTime,
		expireTime:       expireTime,
		refer:            1,
		chunkMergeToken: serverToken + "-" +
			strconv.FormatInt(atomic.AddInt64(&mge.version, 1), 16),
		chunkMergeData: cmData,
		readers:        make([]chan chunkMergeData, 0, 4),
	}

	node.memAddress = fmt.Sprintf("%p", node)
	ocm.m[key] = node
	ocm.expireM.Insert(node)
	occm.insert(node)

	return node.chunkMergeToken
}

func (cm *chunkMerge) delete(key string, chunkMergeToken string, err error) {
	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil || node.chunkMergeToken != chunkMergeToken {
		return
	}

	occm := cm.getOneClientChunkMerge(node.clientToken)

	node.refer--

	node.chunkMergeData.err = err

	for i := 0; i < len(node.readers); i++ {
		node.readers[i] <- node.chunkMergeData
		node.readers[i] = nil
	}
	node.readers = node.readers[:0]

	if err == nil && node.refer > 0 {
		return
	}

	delete(ocm.m, node.key)
	ocm.expireM.Delete(node)
	occm.delete(node)
}

func (cm *chunkMerge) update(key string, chunkMergeToken string,
	cmData chunkMergeData, chunkExpireTimeoutMSec int64) error {

	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil || node.chunkMergeToken != chunkMergeToken {
		return chunkMergeNodeNotFoundErr
	}

	if node.chunkMergeData.mergeDone != "" || node.chunkMergeData.err != nil {
		return chunkMergeNodeAlreadyDoneErr
	}

	if cmData.fragmentIndex != node.fragmentIndex+1 {
		return invalidChunkMergeFragmentIndexErr
	}

	expireTime := time.Now().UnixMilli() + chunkExpireTimeoutMSec
	if node.expireTime < expireTime {
		ocm.expireM.Delete(node)
		node.expireTime = expireTime
		ocm.expireM.Insert(node)
	}

	node.updateExpireTime = expireTime
	node.chunkMergeData = cmData

	for i := 0; i < len(node.readers); i++ {
		node.readers[i] <- cmData
		node.readers[i] = nil
	}
	node.readers = node.readers[:0]

	return nil
}

func (cm *chunkMerge) join(key string, chunkExpireTimeout int64) (chunkMergeToken string, err error) {
	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil {
		return "", chunkMergeNodeNotFoundErr
	}

	node.refer++

	now := time.Now().UnixMilli()
	expireTime := now + chunkExpireTimeout
	// update未过期的情况下, 对node进行续期.
	if node.updateExpireTime > now && node.expireTime < expireTime {
		ocm.expireM.Delete(node)
		node.expireTime = expireTime
		ocm.expireM.Insert(node)
	}

	return node.chunkMergeToken, nil
}

func (cm *chunkMerge) release(key string, chunkMergeToken string) (err error) {
	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil || node.chunkMergeToken != chunkMergeToken {
		return chunkMergeNodeNotFoundErr
	}

	node.refer--

	if node.refer != 0 {
		return nil
	}

	occm := cm.getOneClientChunkMerge(node.clientToken)

	delete(ocm.m, node.key)
	ocm.expireM.Delete(node)
	occm.delete(node)
	return nil
}

func (cm *chunkMerge) getCmData(key string,
	chunkMergeToken string, fragmentIndex int64,
	chunkExpireTimeout int64) (c chan chunkMergeData, cmData chunkMergeData, err error) {
	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil || node.chunkMergeToken != chunkMergeToken {
		return nil, chunkMergeData{}, chunkMergeNodeNotFoundErr
	}

	if fragmentIndex < 1 || fragmentIndex > node.fragmentIndex+1 {
		return nil, chunkMergeData{}, invalidChunkMergeFragmentIndexErr
	}

	now := time.Now().UnixMilli()
	expireTime := now + chunkExpireTimeout
	// update未过期的情况下, 对node进行续期.
	if node.updateExpireTime > now && node.expireTime < expireTime {
		ocm.expireM.Delete(node)
		node.expireTime = expireTime
		ocm.expireM.Insert(node)
	}

	if fragmentIndex == node.fragmentIndex+1 {
		var c = make(chan chunkMergeData, 1)
		node.readers = append(node.readers, c)
		return c, chunkMergeData{}, nil
	}

	return nil, node.chunkMergeData, nil
}

func (cm *chunkMerge) forceDelete(key string, chunkMergeToken string, err error) {
	ocm := cm.getOneChunkMerge(key)

	ocm.Lock()
	defer ocm.Unlock()

	node := ocm.m[key]
	if node == nil || node.chunkMergeToken != chunkMergeToken {
		return
	}

	occm := cm.getOneClientChunkMerge(node.clientToken)

	delete(ocm.m, node.key)
	ocm.expireM.Delete(node)
	occm.delete(node)

	node.chunkMergeData.err = err
	for i := 0; i < len(node.readers); i++ {
		node.readers[i] <- node.chunkMergeData
		node.readers[i] = nil
	}
	node.readers = node.readers[:0]
}

func (cm *chunkMerge) clientAbort(clientToken string) {
	occm := cm.getOneClientChunkMerge(clientToken)

	tmp := make(map[string]*chunkMergeNode)

	occm.Lock()
	ctcm := occm.m[clientToken]
	for k, v := range ctcm {
		tmp[k] = v
	}
	occm.Unlock()

	for k, v := range tmp {
		cm.forceDelete(k, v.chunkMergeToken, chunkMergeClientAbortErr)
	}
}

func (cm *chunkMerge) chunkExpire(i int) {
	ocm := &cm.ocms[i]
	ocm.Lock()
	defer ocm.Unlock()

	for {
		it := ocm.expireM.Min()
		if it == nil {
			return
		}

		now := time.Now().UnixMilli()
		node := it.(*chunkMergeNode)
		if node.expireTime > now {
			return
		}

		ocm.expireM.Delete(node)
		mNode := ocm.m[node.key]
		if mNode == nil || mNode.memAddress != node.memAddress {
			continue
		}

		occm := cm.getOneClientChunkMerge(node.clientToken)
		delete(ocm.m, node.key)
		ocm.expireM.Delete(node)
		occm.delete(node)

		node.chunkMergeData.err = chunkMergeNodeExpireErr
		for i := 0; i < len(node.readers); i++ {
			node.readers[i] <- node.chunkMergeData
			node.readers[i] = nil
		}
		node.readers = node.readers[:0]
	}
}

func (cm *chunkMerge) expireLoop() {
	ticker := time.NewTicker(time.Second)
	for {
		<-ticker.C
		for i := 0; i < slot; i++ {
			cm.chunkExpire(i)
		}
	}
}

func openChunkMerge() {
	cm = &chunkMerge{}
	for i := 0; i < slot; i++ {
		cm.ocms[i].m = make(map[string]*chunkMergeNode)
		cm.ocms[i].expireM = rbtree.New()
		cm.occms[i].m = make(map[string]map[string]*chunkMergeNode)
	}

	go cm.expireLoop()
}
